package com.intct.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.utils.KafkaUtil;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.api.scala.typeutils.Types;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.HashSet;
import java.util.Set;

/**
 * @author gufg
 * @since 2025-09-22 16:06
 */
public class DwsVehicleAPI {
    public static void main(String[] args) throws Exception {
        // 配置环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 开启检查
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // 从工程中resoucres路径读取配置文件
        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(DwsVehicleAPI.class.getResource("/intct.properties").getFile());

        // 数据源配置 ods kafka
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(parameterTool, "dwd_vehicle");
        DataStreamSource<String> sourceDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "dws_vehicle_source");

        /*
            {
              "actionTime": 1758074456,
              "areaId": "110102",
              "code": "110102",
              "driverId": "468",
              "carShelfNumber": "Y893878850D",
              "car": "222",
              "name": "北京市西城区",
              "type": "0",
              "speed": 46
            }
         */

        // 配置水位线
        SingleOutputStreamOperator<String> vechicleDS = sourceDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        // 乱序处理
                        .<String>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                            @Override
                            public long extractTimestamp(String jsonStr, long recordTimestamp) {
                                // 设置事件时间
                                return Long.parseLong(JSON.parseObject(jsonStr).getString("actionTime")) * 1000;
                            }
                        })
        );

        // 1、按照区域进行分组
        KeyedStream<String, String> keyByDS = vechicleDS.keyBy(value -> JSON.parseObject(value).getString("areaId"));

        // 定义侧输出流
        OutputTag<String> vechicleOutputTag = new OutputTag<>("vehcile_out", Types.STRING());

        // 2、窗口
        SingleOutputStreamOperator<String> processDS = keyByDS
                // 窗口类型 -- 滚动
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 推迟关窗时间
                .allowedLateness(Time.seconds(10))
                // 数据窗口关闭后，数据到达处理
                .sideOutputLateData(vechicleOutputTag)
                // 处理数据
                .process(new ProcessWindowFunction<String, String, String, TimeWindow>() {
                    @Override
                    public void process(
                            String key,
                            ProcessWindowFunction<String, String, String, TimeWindow>.Context context,
                            Iterable<String> elements,
                            Collector<String> out) throws Exception {

                        // 窗口开时间
                        String startTime = DateFormatUtils.format(context.window().getStart(), "yyyy-MM-dd HH:mm:ss");
                        // 窗口结束时间
                        String endTime = DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss");

                        // 处理数据  -- 窗口时间内的某个区域的所有车辆信息
                        // 声明Java的Set集合
                        // TODO 集成三方缓存功能 或者 直接采用Redis方式
                        Set vchicleSet = new HashSet<>();

                        // 迭代数据
                        for (String element : elements) {
                            JSONObject jsonObj = JSON.parseObject(element);
                            vchicleSet.add(jsonObj.getString("car"));
                        }

                        JSONObject newJsonObj = new JSONObject(true);
                        newJsonObj.put("area_id", key);
                        newJsonObj.put("num", vchicleSet.size());
                        newJsonObj.put("start_time", startTime);
                        newJsonObj.put("end_time", endTime);

                        out.collect(newJsonObj.toJSONString());

                    }
                });
        processDS.print(">>>>");

        // 4、退维

		// 5、写出数据HBase

        // 启动作业
        env.execute("dws_vehicle");
    }
}
